package com.le.tester.webpage.redis.service.impl;

import com.le.tester.common.Const;
import com.le.tester.delay.BaseDelayed;
import com.le.tester.delay.OnDelayedListener;
import com.le.tester.utils.ThreadPoolUtil;
import com.le.tester.webpage.redis.service.DelayedService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.awt.image.AreaAveragingScaleFilter;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 描述：delayedService服务，启动监听
 */
@Service
public class DelayedServiceImpl implements DelayedService {

    private static final Logger LOG = LoggerFactory.getLogger(DelayedServiceImpl.class);

    //DelayQueue是一个无界的BlockingQueue
    private static DelayQueue<BaseDelayed<?>> delayQueue = new DelayQueue<>();

    //concurrentHashMap,多线程下使用的map
    //我们需要在这个map中修改数据了，读写
    private static ConcurrentHashMap<Class<? extends BaseDelayed>, OnDelayedListener> listeners = new ConcurrentHashMap<>();

    //最具有原子性,可见性，查看状态
    private static AtomicBoolean start = new AtomicBoolean(false);

    /**
     * 在这里可以初始化想要的bean，这里初始化了队列
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        LOG.info("DelayedService afterPropertiesSet当前的时间：{}", LocalDateTime.now().format(DateTimeFormatter.ofPattern(Const.STANDARD_FORMAT)));
        start();
    }

    /**
     * 新注册的监听
     *
     * @param <T>
     */

    public <T extends BaseDelayed<?>> void registerListen(Class<T> clazz, OnDelayedListener<T> listener) {
        if (clazz == null || listener == null) {
            return;
        }
        //通过反射，将CurrentHashMap放进监听
        listeners.put(clazz, listener);
    }


    //拿到具有监听的队列中的实例
    private <T> void start() {
        if (start.get()) {
            return;
        }
        start.set(true);
        new Thread(new Runnable() {
            @SuppressWarnings({"rawtypes", "unchecked"})
            @Override
            public void run() {
                LOG.info("delayedService start");
                try {
                    while (true) {
                        //从队列中把值取出来
                        BaseDelayed<?> delayed = delayQueue.take();
                        //通过反射找到队列的监听
                        OnDelayedListener onDelayedListener = listeners.get(delayed.getClass());
                        //有监听，使用线程池执行
                        if (onDelayedListener != null) {
                            ThreadPoolUtil.execute(new Runnable() {
                                @Override
                                public void run() {
                                    onDelayedListener.onDelayedArrived(delayed);
                                }
                            });
                        } else {
                            LOG.warn("没有找到监听，丢弃：{}", delayed);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }, "delayedService").start();
    }

    //往队列中放东西
    public void add(BaseDelayed<?> baseDelayed) {
        //原子性的判断是不是启动了
        if (!start.get()) {
            throw new RuntimeException("delayedService is not start");
        }
        LOG.info("addDelayedOrder:{}", baseDelayed);
        delayQueue.add(baseDelayed);
    }

    //获取队列
    @SuppressWarnings("unchecked")
    public <T extends BaseDelayed, K> T getDelayed(Class<T> clazz, K key) {
        BaseDelayed<?>[] array = delayQueue.toArray(new BaseDelayed<?>[]{});
        if (array.length <= 0) {
            return null;
        }
        for (BaseDelayed<?> delayed : array) {
            if (delayed.getClass() != clazz) {
                continue;
            }
            if (delayed.getValue().equals(key)) {
                return (T) delayed;
            }
        }
        return null;
    }

    //移除队列中的信息
    @SuppressWarnings("unchecked")
    public <T, D extends BaseDelayed<?>> D remove(Class<T> clazz, T value) {
        if (!start.get()) {
            throw new RuntimeException("delayedService is not start");
        }
        BaseDelayed<?>[] array = delayQueue.toArray(new BaseDelayed<?>[]{});
        if (array.length <= 0) {
            return null;
        }
        D target = null;
        for (BaseDelayed<?> delayed : array) {
            if (delayed.getClass() != clazz) {
                continue;
            }
            if (delayed.getValue().equals(value)) {
                target = (D) delayed.getValue();
                break;
            }

        }
        if (target != null) {
            delayQueue.remove(target);
        }
        return target;

    }

}
